package com.rj.data.task.impl;

import com.google.common.io.Files;
import com.google.common.io.LineProcessor;
import com.rj.data.sink.Sink;
import com.rj.data.source.FileSource;
import com.rj.data.source.Source;
import com.rj.data.task.CallableTask;
import lombok.Builder;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;

/**
 * 用于文件按行的多线程调用处理
 */
public class CallableFileTask<L> extends CallableTask<Boolean> {

    private LineProcessor<L> lineProcessor;

    @Builder(builderMethodName = "callableFileTaskBuilder")
    public CallableFileTask(Source source, List<Sink> sinks, LineProcessor<L> lineProcessor) {
        super(source, sinks);
        this.lineProcessor = lineProcessor;
    }


    @Override
    public Boolean runTask() throws IOException {
        System.out.println("callable process file : " + ((FileSource) getSource()).getAbsolutePath());
        L result = Files.asCharSource((FileSource) getSource(), Charset.forName("UTF-8"))
                .readLines(lineProcessor);
        //遍历输出
        getSinks().forEach(sink -> {
            try {
                sink.writeLines(result, true, true, ",");
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        return true;
    }
}